package dws

import java.{lang, util}

import com.alibaba.fastjson.{JSON, JSONArray, JSONObject}
import modes.{DimAreaInfo, OrderInfo, OrderPriceInfo, RichOrderPriceInfo, TableObject}
import ods.SourceKafka
import org.apache.flink.api.common.functions.{AggregateFunction, MapFunction, RichFlatMapFunction, RichMapFunction}
import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.scala.function.WindowFunction
import org.apache.flink.util.Collector
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.streaming.api.windowing.windows.TimeWindow
import org.apache.flink.table.api.bridge.scala.StreamTableEnvironment

import scala.collection.mutable

object AreaOrderStreaming {
  def main(args: Array[String]): Unit = {
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    val ordersBinlogSrc: DataStream[String] = env.addSource(SourceKafka.getSource("stage9_binlog"))

    val ordersBinlog: DataStream[util.ArrayList[TableObject]] = ordersBinlogSrc
      .filter(x => {
        if (JSON.parseObject(x).getJSONArray("data") != null) {
          true
        } else {
          false
        }
      }).map(x => {
      val json: JSONObject = JSON.parseObject(x)
      val database: String = json.getString("database")
      val table: String = json.getString("table")
      val typeInfo: String = json.getString("type")
      val dataInfos: JSONArray = json.getJSONArray("data")
      val objects = new util.ArrayList[TableObject]()
      dataInfos.forEach(data => {
        objects.add(TableObject(database, table, typeInfo, data.toString))
      })
      objects
    })

    val mapped: DataStream[RichOrderPriceInfo] = ordersBinlog.flatMap(new RichFlatMapFunction[util.ArrayList[TableObject], RichOrderPriceInfo] {
      var areaDimMap: mutable.HashMap[String, (String,String)] = _

      override def open(parameters: Configuration): Unit = {
        areaDimMap = new mutable.HashMap[String, (String,String)]()
        val map: mutable.HashMap[String, String] = new DimAreaHBaseReader().readAll()
        map.foreach(entry => {
          val items = entry._2.split("-")
          areaDimMap.put(entry._1, (items(2), items(4)))
        })
      }

      override def flatMap(value: util.ArrayList[TableObject],out: Collector[RichOrderPriceInfo]): Unit = {
        value.forEach(tableObject => {
          val database: String = tableObject.database
          val table: String = tableObject.tableName
          val typeInfo: String = tableObject.typeInfo
          val dataInfos: String = tableObject.dataInfo
          if ("stage9".equalsIgnoreCase(database)) {
            if ("lagou_trade_orders".equalsIgnoreCase(table)) {
              val info: OrderInfo = JSON.parseObject(dataInfos, classOf[OrderInfo])
              val cityAndProvince: (String, String) = areaDimMap.get(info.areaId).get
              out.collect(RichOrderPriceInfo(info.orderId.toInt, cityAndProvince._1, cityAndProvince._2, info.totalMoney.toDouble))
            }
          }
        })
      }

      override def close(): Unit = super.close()
    })

    val keyed: KeyedStream[RichOrderPriceInfo, String] = mapped.keyBy(x => x.city + "-" + x.province)
    val windowed: WindowedStream[RichOrderPriceInfo, String, TimeWindow] = keyed.timeWindow(Time.seconds(60 * 60), Time.seconds(10))
    val result: DataStream[(String, (Int, Double))] = windowed.aggregate(new AggregateFunction[RichOrderPriceInfo, (Int, Double), (Int, Double)] {
      override def createAccumulator(): (Int, Double) = {
        (0, 0.0)
      }

      override def add(value: RichOrderPriceInfo, accumulator: (Int, Double)): (Int, Double) = {
        (accumulator._1 + 1, accumulator._2 + value.totalMoney)
      }

      override def getResult(accumulator: (Int, Double)): (Int, Double) = {
        accumulator
      }

      override def merge(a: (Int, Double), b: (Int, Double)): (Int, Double) = ???
    }, new WindowFunction[(Int, Double), (String, (Int, Double)), String, TimeWindow] {

      override def apply(key: String, window: TimeWindow, input: Iterable[(Int, Double)], out: Collector[(String, (Int, Double))]): Unit = {
        out.collect((key, input.iterator.next()))
      }
    })

    result.print()

    env.execute()
  }
}
